package com.simple.datax.loader.impl;

import java.util.ArrayList;
import java.util.List;

import javax.jms.MessageListener;

import org.apache.activemq.ActiveMQConnection;

import com.alibaba.dubbo.common.logger.Logger;
import com.alibaba.dubbo.common.logger.LoggerFactory;
import com.simple.datax.api.Loader;
import com.simple.datax.conn.BaseConnector;
import com.simple.datax.conn.ConnectorPool;
import com.simple.datax.conn.impl.ActiveMQConnImpl;
import com.simple.datax.conn.impl.ConnectorPoolManager;
import com.simple.datax.conn.impl.MQBaseConnector;
import com.simple.datax.core.model.connpara.CommonConnPara;
import com.simple.datax.recv.impl.ActiveMQMessageListener;

public class DBMQLoaderImpl implements Loader {
	private static final Logger logger = LoggerFactory
			.getLogger(DBMQLoaderImpl.class);

	private ConnectorPool connectorPool = ConnectorPoolManager.getInstance().getConnectorPool(ConnectorPoolManager.CONN_RECV, null, null);
	private MessageListener listener;
	
	@Override
	public void run() {
		// Get MQ connection information and initial all MQ 
		List<CommonConnPara> paras = loadAllConnectorInfo();
		BaseConnector connector = null;
		for (CommonConnPara para:paras) {
			// get connector from pool
			try {
				connector = connectorPool.borrowObject(para);
				
				// got connector then continue
				if (connector != null) {
					logger.debug("Initialized MQ Connector successfully.");
					if (connector instanceof MQBaseConnector) {
						((ActiveMQMessageListener) listener).setConnPara(para);
						((MQBaseConnector) connector).setMessageListener(listener);
					} else {
						logger.error("Connector got from pool is not MQBaseConnector class.");
					}
				} else {
					logger.debug("Initialized MQ Connector failed.");
				}
			} catch (Exception e) {
				logger.error("Reciver get message error.", e);
			}
		}
	}
	
	private List<CommonConnPara> loadAllConnectorInfo() {
		List<CommonConnPara> ret = new ArrayList<CommonConnPara>(1000);
		CommonConnPara para = null;
		
//		for (int i=0; i<100; i++) {
			for (int j=1; j<2; j++) {
				para = new CommonConnPara();
				ret.add(para);
				para.setTransportType(ActiveMQConnImpl.TYPE);
				//"URL$USER$PASSWORD$SUBJECT";
				para.setTransportPara("tcp://JoeJinWin7:61616$"+ActiveMQConnection.DEFAULT_USER+"$"+ActiveMQConnection.DEFAULT_PASSWORD+"$QUEUE_"+j);
			}
//		}
		return ret;
	}

	public MessageListener getListener() {
		return listener;
	}

	public void setListener(MessageListener listener) {
		this.listener = listener;
	}

}
